import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
import java.util.Random;

/**
 * @Time ： 2024/10/8
 * @Auth ： cheng
 * @IDE ：ide
 * @Motto：ABC(Always Be Coding)
 **/
public class KafkaPro {
    public static void main(String[] args)  {

// 链接 centos kafka，需要kafka 客户端 ，需要 maven 添加 kafka 的依赖
        // 编码： 1） 构建 kafka 的配置信息 kafka的ip地址，map  key，value ---> 序列化 ---》 kafka的服务
        //        创建一个producer ， 发送 随机数据，关闭 producer

        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.10.130:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        while (true){

            KafkaProducer kafkaProducer = new KafkaProducer<String,String>(properties);

            try {


            for (int i = 0; i < 5; i++) {
                Random random = new Random();
                String[] categoryArray = new String[]{"小明","小米","小兰","小张","小雪"};
                int rd = random.nextInt(5);
                String mockData = categoryArray[rd];
                System.out.println("data is " + mockData);
                kafkaProducer.send(new ProducerRecord<String,String>("16testnew",mockData));

            }
            try {
                Thread.sleep(2000);
            }catch (Exception e){
                e.printStackTrace();
            }


            kafkaProducer.close();

            }catch (Exception e){
                e.printStackTrace();
            }
        }







    }
}
